# Decompiled with PyLingual (https://pylingual.io)
# Internal filename: 密码开锁_推送识别版_打包版.py
# Bytecode version: 3.12.0rc2 (3531)
# Source timestamp: 1970-01-01 00:00:00 UTC (0)

import time
import os
import sys
import threading
import logging
import random
from datetime import datetime
import uiautomator2 as u2
import openpyxl
import serial
app_pgname = 'com.xiaomi.smarthome'
file_name_MCU = datetime.now().strftime('%Y%m%d%H%M') + '_MCU.log'
title_list = ['测试次数', '进入插件时间', '开锁测试结果', '异常备注（密码开锁）', '消息推送结果']

class PasswordUnlocking(object):

    def __init__(self):
        self.d = u2.connect(devices_id)
        self.ser = PasswordUnlocking.serial_port()
        self.log = open(os.path.dirname(sys.argv[0]) + '/' + file_name_MCU, 'a', encoding='utf-8', errors='replace')
        self.running = True
        self.__create_excel()
        self.times = None
        self.pass_times = 0
        self.fail_times = 0
        self.block_times = 0
        self.result = None
        self.error_log = None
        self.start_time = None
        self.result_msg = None
        self.fail_times_msg = 0
        self.logger = PasswordUnlocking.__create_file()

    @staticmethod
    def __create_file():
        logger = logging.getLogger(__name__)
        logger.setLevel(logging.DEBUG)
        log_format = '%(asctime)s - %(levelname)s - %(message)s'
        file_name = datetime.now().strftime('%Y%m%d%H%M') + '密码开锁控制台.log'
        file_handler = logging.FileHandler(os.path.dirname(sys.argv[0]) + '/' + file_name)
        file_handler.setFormatter(logging.Formatter(log_format))
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(logging.Formatter(log_format))
        logger.addHandler(file_handler)
        logger.addHandler(console_handler)
        return logger

    def __create_excel(self):
        """创建数据表格和截图文件夹"""
        self.excel_path = os.path.dirname(sys.argv[0]) + '/密码开锁测试数据' + str(PasswordUnlocking.__get_time1(datetime.now())) + '.xlsx'
        self.picture_path = os.path.dirname(sys.argv[0]) + '/密码开锁异常截图' + str(PasswordUnlocking.__get_time1(datetime.now()))
        self.picture_path1 = str(self.picture_path) + '/其它异常'
        os.makedirs(self.picture_path, exist_ok=True)
        os.makedirs(self.picture_path1, exist_ok=True)
        excel_makedirs = openpyxl.Workbook()
        excel_makedirs.save(self.excel_path)
        self.excel = openpyxl.load_workbook(self.excel_path)
        self.excel_sheet = self.excel.active
        self.excel_sheet.auto_filter.ref = 'A1:E1'
        self.excel_sheet.append(title_list)
        self.excel_sheet['K1'] = '测试总次数'
        self.excel_sheet['L1'] = '开锁成功次数'
        self.excel_sheet['M1'] = '开锁失败次数'
        self.excel_sheet['N1'] = '开锁成功率'
        self.excel_sheet['O1'] = '开锁异常次数'
        self.excel_sheet['P1'] = '消息推送失败次数'
        self.excel.save(self.excel_path)

    @staticmethod
    def __get_time(get_time):
        hms = get_time.strftime('%H:%M:%S')
        return hms

    @staticmethod
    def __get_time1(get_time):
        ymdhms = get_time.strftime('%Y%m%d%H%M%S')
        return ymdhms

    @staticmethod
    def __get_time2(get_time):
        hms = get_time.strftime('%H%M%S')
        return hms

    @staticmethod
    def __count_time(end_time, st_time):
        use_time = (end_time - st_time).total_seconds()
        return float(use_time)

    @staticmethod
    def serial_port():
        ser = serial.Serial(port=port_mcu, baudrate=115200, timeout=1)
        if not ser or not ser.is_open:
            ser.open()
        return ser

    def send_to_serial(self, command):
        """前板ser1发送命令到串口"""
        try:
            byte_data = bytes.fromhex(command)
            self.ser.write(byte_data)
        except Exception as e:
            print(f'Error sending command: {str(e)}')

    def read_from_serial(self):
        """前板持续读取串口数据并写入日志文件"""
        while self.running:
            try:
                if self.ser.in_waiting > 0:
                    data = self.ser.read(self.ser.in_waiting)
                    decoded_data = data.decode('utf-8', errors='replace').strip()
                    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
                    self.log.write(f'[{timestamp}] {decoded_data}\n')
                    self.log.flush()
                    time.sleep(0.01)
            except Exception as e:
                pass
            self.logger.info(f'Error reading serial: {str(e)}')

    def app_in(self):
        self.d.app_stop(app_pgname)
        self.d.app_start(app_pgname, use_monkey=True, wait=True)
        if self.d(text=lock_name).exists(timeout=15):
            self.d(text=lock_name).click()
            time.sleep(2)
        else:
            self.logger.info('没有找到指定门锁元素')

    def code_in(self):
        read_thread = threading.Thread(target=self.read_from_serial)
        read_thread.start()
        num = '0F DE 00 81 00 09 00 31 3' + str(pass_word[0]) + ' 3' + str(pass_word[1]) + ' 3' + str(pass_word[2]) + ' 3' + str(pass_word[3]) + ' 3' + str(pass_word[4]) + ' 3' + str(pass_word[5]) + ' 23 01 04'
        for times in range(test_cycles):
            try:
                wait_time = random.randint(wait_time_a, wait_time_b)
                self.error_log = ''
                self.result_msg = ''
                self.result = ''
                self.times = times
                self.logger.info('********************开始第%s轮测试，时间：%s********************' % (self.times + 1, PasswordUnlocking.__get_time(datetime.now())))
                self.start_time = datetime.now()
                self.d.app_stop(app_pgname)
                time.sleep(1)
                self.d.press('back')
                self.logger.info('按返回键，防止桌面异常')
                self.d.app_start(app_pgname, use_monkey=True, wait=True)
                if self.d(text=lock_name).exists(timeout=20):
                    self.d(text=lock_name).click()
                    self.logger.info('点击进入插件')
                    if self.d(text='安全守护中').exists(timeout=60):
                        self.logger.info('识别到元素：安全守护中')
                        time.sleep(1)
                        self.send_to_serial(num)
                        self.logger.info('发送密码开锁指令')
                        if self.d(text='已开锁').exists(timeout=30):
                            self.logger.info('识别到元素：已开锁')
                            self.result = 'Pass'
                            self.pass_times += 1
                            self.logger.info('检测到元素：已开锁，开始下拉左侧栏，检测消息推送')
                            time.sleep(1)
                            self.d.swipe(10, 0, 10, 500, duration=0.2)
                            self.logger.info('下滑左侧通知栏')
                            time.sleep(2)
                            if self.d(textContains='自动上锁').exists(timeout=40):
                                self.logger.info('识别到模糊元素：自动上锁，本轮测试结果：Pass')
                                self.result_msg = 'Pass'
                                time.sleep(1)
                                self.d.click(clear_x, clear_y)
                                self.logger.info('点击X清空消息，返回插件页面')
                                self.logger.info('等待%s秒后开始下一轮测试' % wait_time)
                                self.handle_error(1)
                                time.sleep(wait_time)
                            else:
                                self.logger.info('40秒未检测到元素：自动上锁，本轮测试结果：Fail')
                                self.result_msg = 'Fail'
                                self.fail_times_msg += 1
                                self.handle_error(2)
                                self.d.press('back')
                                self.logger.info('按返回键，返回插件页面')
                                self.logger.info('等待%s秒后开始下一轮测试' % wait_time)
                                time.sleep(wait_time)
                        else:
                            self.result = 'Fail'
                            self.fail_times += 1
                            self.error_log = '未检测到元素：已开锁'
                            self.handle_error(2)
                            print('30秒未检测到元素：已开锁，本轮测试结果：Fail')
                            print('等待%s秒后开始下一轮测试' % wait_time)
                            time.sleep(wait_time)
                    else:
                        self.result = 'Block'
                        self.block_times += 1
                        self.error_log = '未检测到元素：安全守护中'
                        self.logger.info('60秒未检测到元素：安全守护中，本轮测试结果：block,等待%s秒后开始下一轮测试' % wait_time)
                        self.handle_error(2)
                        time.sleep(wait_time)
                else:
                    self.result = 'Block'
                    self.block_times += 1
                    self.error_log = '未检测到门锁'
                    self.logger.info('20秒未找到指定门锁，本轮测试结果：block,等待%s秒后开始下一轮测试' % wait_time)
                    self.handle_error(2)
                    time.sleep(wait_time)
            except Exception as e:
                pass
        else:
            self.running = False
            self.logger.info(f'测试流程异常: {str(e)}')
            self.d.screenshot(self.picture_path1 + '/第%s次测试异常%s.png' % (str(self.times + 1), str(PasswordUnlocking.__get_time2(datetime.now()))))
            self.logger.info('本轮测试异常，跳过本轮测试,等待10秒后开始下一轮测试')
            self.d.app_stop(app_pgname)
            time.sleep(10)

    def excel_append(self, times, start_time, result, error_log, result_msg):
        log_list = [times, start_time, result, error_log, result_msg]
        self.excel_sheet.append(log_list)
        self.excel_sheet['K2'] = self.times + 1
        self.excel_sheet['L2'] = self.pass_times
        self.excel_sheet['M2'] = self.fail_times
        self.excel_sheet['N2'] = '=(L2/(L2+M2))'
        self.excel_sheet['O2'] = self.block_times
        self.excel_sheet['P2'] = self.fail_times_msg
        self.excel_sheet['N2'].number_format = '0.00%'
        self.excel.save(self.excel_path)

    def handle_error(self, state):
        if state == 1:
            self.excel_append(self.times + 1, PasswordUnlocking.__get_time(self.start_time), self.result, self.error_log, self.result_msg)
            self.d.app_stop(app_pgname)
        elif state == 2:
            self.d.screenshot(self.picture_path + '/第%s次测试异常%s.png' % (str(self.times + 1), str(PasswordUnlocking.__get_time2(datetime.now()))))
            self.excel_append(self.times + 1, PasswordUnlocking.__get_time(self.start_time), self.result, self.error_log, self.result_msg)
            self.d.app_stop(app_pgname)
        elif state == 3:
            self.d.screenshot(self.picture_path + '/第%s次测试异常%s.png' % (str(self.times + 1), str(PasswordUnlocking.__get_time2(datetime.now()))))
            self.excel_append(self.times + 1, PasswordUnlocking.__get_time(self.start_time), self.result, self.error_log, self.result_msg)
            self.d.app_stop(app_pgname)
if __name__ == '__main__':
    test_cycles = int(input('输入测试次数:'))
    devices_id = input('输入手机设备id，cmd中输入adb devices (仅连接一台设备时可不填):')
    port_mcu = input('输入MCU串口号(如：COM10):')
    lock_name = input('请输入门锁在APP首页的全称，并确保APP首页能找到该门锁：')
    pass_word = input('请输入6位开锁密码，并确保该密码已添加并能成功开锁：')
    print('输入随机等待时长的最短最长时间，若不需要随机，可输入两个相同整数')
    wait_time_a = int(input('输入一轮结束后的最短等待时间'))
    wait_time_b = int(input('输入一轮结束后的最长等待时间'))
    print('手机消息通知页面x（清除消息按键）的位置（可在开发者选项-指针位置，打开后可看到该位置坐标）')
    clear_x = int(input('输入清除消息按键的X坐标：'))
    clear_y = int(input('输入清除消息按键的Y坐标：'))
    a = PasswordUnlocking()
    a.code_in()